package com.doit.day03;

import com.alibaba.fastjson.JSON;
import lombok.*;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ProducerEventLog {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //开启幂等性
        props.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,"true");
        //开启幂等性的三个参数
        props.setProperty(ProducerConfig.ACKS_CONFIG,"-1");
        props.setProperty(ProducerConfig.RETRIES_CONFIG,"3");
        props.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,"4");

        props.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"abc");

        //搞一个生产者出来==》创建一个topic
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);


        //开启事务
        producer.beginTransaction();

        //提交事务
        producer.commitTransaction();

        //放弃事务
        producer.abortTransaction();


        EventLog eventLog = new EventLog();
        while (true){
            //造数据==》怎么造 创建一个javaBean 然后再将对象转换成json串
            eventLog.setGuid(RandomUtils.nextInt(1,100000));
            eventLog.setEventId(RandomStringUtils.randomAlphabetic(10));
            eventLog.setTimestamp(System.currentTimeMillis());
            //将对象转换成json串
            String json = JSON.toJSONString(eventLog);
            //如何去造数据
            ProducerRecord<String, String> record = new ProducerRecord<>("event-log", json);
            //发送数据
            producer.send(record);
            producer.flush();
            Thread.sleep(RandomUtils.nextInt(10,100));
        }
    }
}

@ToString
@AllArgsConstructor
@NoArgsConstructor
@Setter
@Getter
class EventLog{
    private int guid;
    private String EventId;
    private long timestamp;
    private int isNew;
}


